package org.databandtech.mysql2hive.sink;

import java.sql.*;

import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import java.sql.PreparedStatement;


public class HiveSink extends RichSinkFunction<Tuple5<String,Integer,String,String,String>> {

	private static final long serialVersionUID = 1L;
	private PreparedStatement state ;
    private Connection conn ;
    private String QUERYSQL = "";
    private static String URL = "jdbc:hive2://192.168.13.52:10000/default";
    private static String USER = "root";  // 重要！此处必须填入具有HDFS写入权限的用户名，比如hive文件夹的owner
    private static String PASSWORD = "";  
    
    public HiveSink(String sql) {   	
    	this.QUERYSQL = sql;
    }
    
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        conn = getConnection();
        state = conn.prepareStatement(QUERYSQL);
    }

    @Override
    public void close() throws Exception {
        super.close();
        if (state != null) {
            state.close();
        }
        if (conn != null) {
            conn.close();
        }

    }

    @Override
    public void invoke(Tuple5<String,Integer,String,String,String> value, Context context) throws Exception {
        state.setString(1,value.f0);
        state.setInt(2,value.f1);
        state.setString(3,value.f2);
        state.setString(4,value.f3);
        state.setString(5,value.f4);

        state.executeUpdate();
    }

    private static Connection getConnection() {
        Connection conn = null;
        try {
            String jdbc = "org.apache.hive.jdbc.HiveDriver";
            Class.forName(jdbc);
            //conn = DriverManager.getConnection(URL);
            conn = DriverManager.getConnection(URL, USER, PASSWORD);

        } catch (Exception e) {
            System.out.println(e.getMessage());
        }
        return conn;
    }


}
